package com.liry.zk;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;

/**
 * 分布式锁 - 非公平锁实现
 *
 * @author ALI
 * @since 2022/12/25
 */
@Slf4j
public class NonfairSyncLock {

    private static final String CONNECT_STR = "192.168.17.128:2181";

    private static final int TIME_OUT = 30000;

    private static final String LOCK_PATH = "/lock";

    private static ZooKeeper zookeeper = null;

    /**
     * 获取客户端
     */
    public static ZooKeeper getClient() throws IOException, InterruptedException {
        synchronized (LOCK_PATH) {
            final CountDownLatch latch = new CountDownLatch(1);
            if (zookeeper == null) {
                Watcher startWatcher = watchedEvent -> {
                    if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {
                        log.info("与zookeeper建立连接");
                        latch.countDown();
                    }
                };
                zookeeper = new ZooKeeper(CONNECT_STR, TIME_OUT, startWatcher);
                latch.await();
            } else if (zookeeper.getState() != ZooKeeper.States.CONNECTED) {
                latch.await();
            }
        }
        return zookeeper;
    }

    /**
     * 加锁
     */
    public void lock() {
        while (true) {
            if (tryLock()) {
                return;
            }
            // 加锁失败，增加监听，然后阻塞
            try {
                watch();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 解锁
     */
    public void unlock() {
        try {
            getClient().delete(LOCK_PATH, -1);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 尝试加锁
     */
    private boolean tryLock() {
        try {
            getClient().create(LOCK_PATH, "lock".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                               CreateMode.EPHEMERAL);
        } catch (Exception e) {
            log.error("加锁失败");
            return false;
        }
        return true;
    }

    /**
     * 监听锁节点
     * 当节点存在时，线程阻塞，当节点不存在，直接退出监听
     */
    private void watch() throws InterruptedException, KeeperException, IOException {
        CountDownLatch latch = new CountDownLatch(1);
        Watcher dataWatch = event -> {
            if (event.getType() == Watcher.Event.EventType.NodeDeleted) {
                latch.countDown();
            }
        };
        try {
            getClient().getData(LOCK_PATH, dataWatch, null);
        } catch (KeeperException.NoNodeException e) {
            // 当创建监听时，节点不存在，说明有线程解锁了，那么直接退出，监听步骤，去争抢锁
            return;
        }
        latch.await();
    }

}
